package org.example.kpi;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * KPIIP类包含处理IP地址的MapReduce作业的定义。
 */
public class KPIIP {

    /**
     * KPIIPMapper是映射器类，负责从输入日志中提取URL和IP地址。
     */
	public static class KPIIPMapper extends Mapper<Object, Text, Text, Text> {
		private Text urlText = new Text();
		private Text ipText = new Text();

        /**
         * map方法处理输入的日志行，提取URL和IP地址，并将它们作为键值对输出。
         *
         * @param key 输入日志行的键，此处未使用。
         * @param value 输入日志行的值，预期为文本格式。
         * @param context 上下文对象，用于输出键值对到下一阶段的处理。
         * @throws IOException 如果发生I/O错误。
         * @throws InterruptedException 如果执行过程中线程被中断。
         */
		@Override
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			// 从日志行中过滤IP地址，提取KPI信息
			KPI kpi = KPI.filterIPs(value.toString());
			// 检查KPI信息是否有效
			if (kpi.isValid()) {
				// 设置URL和IP地址作为输出的键值对
				urlText.set(kpi.getRequest());
				ipText.set(kpi.getRemote_addr());
				// 输出键值对到上下文
				context.write(urlText, ipText);
			}
		}

	}

    /**
     * KPIIPReducer是化简器类，负责计算每个URL的唯一IP地址数量。
     */
	public static class KPIIPReducer extends Reducer<Text, Text, Text, Text> {
		private Text result = new Text();

        /**
         * reduce方法接收相同URL的所有IP地址，计算并输出唯一IP地址的数量。
         *
         * @param key 表示URL的文本键，为MapReduce作业中的键值对的一半。
         * @param ips 表示与给定URL相关联的所有IP地址的可迭代集合。
         * @param context 用于输出键值对到结果集合的上下文对象。
         * @throws IOException 当发生I/O错误时抛出。
         * @throws InterruptedException 当执行被中断时抛出。
         */
		@Override
		public void reduce(Text key, Iterable<Text> ips, Context context) throws IOException, InterruptedException {
			Set<Text> ipsSet = new HashSet<Text>(); // 使用HashSet存储唯一IP地址。

			// 遍历ips集合，将所有IP地址添加到ipsSet中。
			// 这段Java代码的作用是将一个ips集合中的所有元素添加到ipsSet集合中。
			// 代码使用了迭代器Iterator<Text>遍历ips集合，通过i.hasNext()判断是否还有下一个元素，
			// 如果有则使用i.next()获取该元素，并将其添加到ipsSet集合中。
			for (Iterator<Text> i = ips.iterator(); i.hasNext();) {
				ipsSet.add(i.next());
			}

			// 输出URL和唯一IP地址的数量。
			context.write(key, new Text(ipsSet.size() + ""));
		}

	}

    /**
     * 主函数用于配置并提交KPIIP作业。
     * 该作业会处理输入路径的Web日志数据，计算IP相关的KPI指标，并将结果存储到输出路径。
     *
     * @param args 命令行参数，包括输入路径和输出路径。格式为["输入路径", "输出路径"]。
     * @throws Exception 如果作业提交或执行失败，则抛出异常。
     */
	public static void main(String[] args) throws Exception {
		// 默认输入输出路径
		String input = "hdfs://master:9000/weblog/logfile";
		String output = "hdfs://master:9000/weblog/kpi/ip";

		// 配置作业的初始设置
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "KPIIP");
		job.setJarByClass(KPIIP.class); // 指定作业所使用的JAR包

		// 配置MapReduce的输出格式
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		// 配置最终输出的键值对类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		// 指定Mapper和Reducer类
		job.setMapperClass(KPIIPMapper.class);
		// job.setCombinerClass(KPIIPReducer.class); // 注释掉的代码用于合并Map阶段的输出
		job.setReducerClass(KPIIPReducer.class); // 指定Reducer类

		// 设置输入输出路径
		FileInputFormat.addInputPath(job, new Path(input));
		FileOutputFormat.setOutputPath(job, new Path(output));

		// 提交作业并等待完成
		if (!job.waitForCompletion(true))
			return; // 如果作业未成功完成，则退出程序
	}


}
